package com.lm.activiti.boot.mq;

import com.lm.activiti.boot.mq.common.ConnectionUtils;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeoutException;

/**
 * @author lm_jcf
 * @date 2020/7/15 15:40
 */
public class RabbitMq {
    // 默认发送，直接将消息发送到某个队列，默认交换机类型 exchange Type = direct

    public static void main(String[] args) throws IOException, TimeoutException {
        RabbitMq producer = new RabbitMq();

        producer.read();
        producer.read_2();

        producer.producer();
    }

    @Test
    public void producer() throws IOException, TimeoutException {

        Connection connection = ConnectionUtils.getConnection();

        String queueName = "test_direct_queue";

        // 从链接中获取一个通道
        Channel channel = connection.createChannel();

        // 创建队列声明
        channel.queueDeclare(queueName, false, false, false, null);

        /**
         * TODO 消息持久化：queueDeclare.durable = true 对于已存在的queue 不可声明不一样参数的queue
         */

        channel.basicPublish("", queueName, null, UUID.randomUUID().toString().getBytes());

        /**
         * TODO 公平分发（不同消费者处理消息效能不一样，公平分发能优先处理快的消费者消费）：设置为：成功应答条数才发送消息给消费者
         *
         */
        int prefetchCount = 1;
        channel.basicQos(prefetchCount);

        channel.close();
        connection.close();
    }

    /**
     * 使用交换机
     *
     * exchange 交换机：接收生产者消息，推送消息给队列
     *
     * fanout exchange：不处理路由键，自动把消息转发到绑定的所有队列
     * direct exchange：处理路由键，自动把消息转发到绑定的指定key队列，通过routingKey 指定队列
     *
     */
    @Test
    public void producer_exchange() throws IOException, TimeoutException {

        Connection connection = ConnectionUtils.getConnection();

        String queueName = "test_exchange_fanout_queue";
        String exchangeName = "test_exchange_fanout";

        // 从链接中获取一个通道
        Channel channel = connection.createChannel();

        // 创建交换机
        channel.exchangeDeclare(exchangeName, "fanout");

        /**
         * TODO fanout exchange：不处理路由键，把消息转发到绑定的所有队列
         * TODO direct exchange：指定处理路由键，把消息转发到绑定的指定key队列，通过routingKey 指定队列
         */

        channel.basicPublish(exchangeName, "only", null, "exchange_1".getBytes());

        channel.close();
        connection.close();
    }

    /**
     * 使用交换机：消费者
     */
    @Test
    public void consumer_exchange() throws IOException, TimeoutException {

        Connection connection = ConnectionUtils.getConnection();

        String queueName = "test_exchange_fanout_queue";
        String exchangeName = "test_exchange_fanout";

        // 从链接中获取一个通道
        Channel channel = connection.createChannel();

        // 队列声明
        channel.queueDeclare(queueName, false, false, false, null);

        // 绑定队列到交换机上
        channel.queueBind(queueName, exchangeName, "only");

        // 定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {

            // 获取到达的消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String msg = new String(body, "utf-8");
                System.out.println("read_1 msg：" + msg);

//                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };

        boolean autoAck = true; // 自动应答
        // 绑定队列与消费者
        channel.basicConsume(queueName, autoAck, consumer);

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


        channel.close();
        connection.close();
    }


    private static String queueName = "test_direct_queue";

    @Test
    public void read() throws IOException, TimeoutException {

        Connection connection = ConnectionUtils.getConnection();

        // 从链接中获取一个通道
        Channel channel = connection.createChannel();

        // 队列声明
        channel.queueDeclare(queueName, false, false, false, null);

        // 保证一次只分发一条消息
        channel.basicQos(1);

        // 定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {

            // 获取到达的消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String msg = new String(body, "utf-8");
                System.out.println("read_1 msg：" + msg);

                channel.basicAck(envelope.getDeliveryTag(), false);

            }
        };

        /**
         * TODO autoAck = true 自动应答，一旦rabbitmq将消息分发给消费者，就会从内存删除该消息，当目标A消费者接收消息前处理错误，则会丢失该消息
         * TODO autoAck = false（默认） 手动应答，如若rabbitmq未收到消息回执，则会再次把消息发送给其他相同消费者
         * 当rabbitmq挂掉，消息一样会丢失，可做持久化操作
         *
         *
         */
        boolean autoAck = false; // 自动应答
        // 绑定队列
        channel.basicConsume(queueName, autoAck, consumer);

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void read_2() throws IOException, TimeoutException {

        Connection connection = ConnectionUtils.getConnection();

        // 从链接中获取一个通道
        Channel channel = connection.createChannel();

        // 队列声明
        channel.queueDeclare(queueName, false, false, false, null);

        // 定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {

            // 获取到达的消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String msg = new String(body, "utf-8");
                System.out.println("read_2 msg：" + msg);
            }
        };
        // 绑定队列
        channel.basicConsume(queueName, true, consumer);
    }

    /**
     *
     * 交换机类型：exchange type = topic 主题模式
     * 通过通配符匹配与之绑定的队列
     * 将 routing key 和某个模式匹配，
     * # 匹配一个或多个、* 匹配一个
     */

    @Test
    public void producer_exchange_topic() throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();

        String queueName = "test_exchange_topic_queue";
        String exchangeName = "test_exchange_topic";

        // 从链接中获取一个通道
        Channel channel = connection.createChannel();

        // 创建交换机
        channel.exchangeDeclare(exchangeName, "topic");

        channel.basicPublish(exchangeName, "goods.update", null, "exchange_topic_1".getBytes());

        channel.close();
        connection.close();
    }

    /**
     * 消息事务机制
     *  TODO AMQP 事务机制
     *  通道为 AMQP 模式
     *  txSelect txCommit txRollback
     *
     *  txSelect 将当前 channel 设置成 Transactional 模式
     *  txCommit 提交事务
     *  txRollback 事务回滚
     *
     * @throws IOException
     * @throws TimeoutException
     */
    @Test
    public void producer_exchange_tx() throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();

        String queueName = "test_exchange_topic_queue";
        String exchangeName = "test_exchange_topic";

        // 从链接中获取一个通道
        Channel channel = connection.createChannel();

        channel.queueDeclare(queueName, false, false, false, null);

        try {
            // 开启事务
            channel.txSelect();

            channel.basicPublish(exchangeName, "goods.update", null, "exchange_topic_1".getBytes());

            // 提交事务
            channel.txCommit();
        } catch (Exception e) {
            // 事务回滚
            channel.txRollback();
            System.out.println("txRollback");
        }

        channel.close();
        connection.close();
    }

    /**
     * TODO 消息事务机制 Confirm 模式 同步
     * 通道 为confirm 模式，一旦信道进入confirm模式，所有在该信道上发布的消息都会指派一个唯一的 ID（从1开始），一旦消息被投递到所有匹配的队列后，
     * broker 就会发送一个确认信息给生产者（包含唯一的ID）以告知生产者消息填入队列成功，若队列设为持久化的，则在成功持久化后才会返回回执，
     * broker 回传的确认消息中 deliver-tag 域包含了确认消息的序列号，此外 broker 还可以设置basic.ack 的multiple域，表示到这个序列号之前的消息
     * 都已得到了处理。
     *
     * confirm 模式为异步的，
     *
     * 通过 waitForConfirms() 得到此次操作的回执
     */
    @Test
    public void producer_exchange_tx_confirm() throws IOException, TimeoutException, InterruptedException {
        Connection connection = ConnectionUtils.getConnection();

        String queueName = "test_confirm_queue";

        // 从链接中获取一个通道
        Channel channel = connection.createChannel();

        channel.queueDeclare(queueName, false, false, false, null);

        // 生产者调用 confirmSelect 将channel 设置为confirm模式
        channel.confirmSelect();

        channel.basicPublish("", queueName, null, "exchange_topic_1".getBytes());

        // 确认
        if (!channel.waitForConfirms()) {
            System.out.println("message send failed");
        } else {
            System.out.println("mssage send ok");
        }

        channel.close();
        connection.close();
    }

    /**
     * TODO 消息事务机制 Confirm 模式 异步
     * Channel 对象提供了ConfirmListener() 回调方法只包含 deliveryTag（当前Channel 发出的消息序号），我们需要自己为每一个 Channel 维护一个
     * unconfirm 的消息序号集合，每次 publish 集合元素加 1，每次回调 handleAck 方法，unconfirm方法删掉对应的一条（multiple=false）或多条
     * （multiple = true）记录，运行效率高的集合为 有序集合 SortedSet
     *
     * @throws IOException
     * @throws TimeoutException
     * @throws InterruptedException
     */
    @Test
    public void producer_exchange_tx_confirm_async() throws IOException, TimeoutException, InterruptedException {
        Connection connection = ConnectionUtils.getConnection();

        String queueName = "test_confirm_queue_async";

        // 从链接中获取一个通道
        Channel channel = connection.createChannel();

        channel.queueDeclare(queueName, false, false, false, null);

        // 生产者调用 confirmSelect 将channel 设置为confirm模式
        channel.confirmSelect();

        // 未确认的消息ID标识
        final SortedSet<Long> unconfirmMsgSet = Collections.synchronizedSortedSet(new TreeSet<Long>());

        // 通道添加监听
        channel.addConfirmListener(new ConfirmListener() {
            //
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {

                if (multiple) {
                    System.out.println("---handleAck --- multiple");
                    unconfirmMsgSet.headSet(deliveryTag + 1).clear();
                } else {
                    System.out.println("---handleAck --- multiple false");
                    unconfirmMsgSet.remove(deliveryTag);
                }
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if (multiple) {
                    System.out.println("---handleNack --- multiple");
                    unconfirmMsgSet.headSet(deliveryTag + 1).clear();
                } else {
                    System.out.println("---handleNack --- multiple false");
                    unconfirmMsgSet.remove(deliveryTag);
                }
            }
        });

        for (int i = 0; i < 10; i++) {

            // 获取每次 publish 的消息序号
            long seqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("", queueName, null, "exchange_topic_1".getBytes());
            unconfirmMsgSet.add(seqNo);
        }

        channel.close();
        connection.close();
    }

    @Test
    public void consumer_exchange_topic() throws IOException, TimeoutException, InterruptedException {

        Connection connection = ConnectionUtils.getConnection();

        Channel channel = connection.createChannel();

//        String queueName = "test_exchange_topic_queue";
        String queueName = "test_confirm_queue";
        String exchangeName = "test_exchange_topic";

        // 队列声明
        channel.queueDeclare(queueName, false, false, false, null);

        channel.queueBind(queueName, exchangeName, "goods.update");

        // 保证一次只分发一条消息
        channel.basicQos(1);

        // 定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {

            // 获取到达的消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String msg = new String(body, "utf-8");
                System.out.println("read_1 msg：" + msg);

                channel.basicAck(envelope.getDeliveryTag(), false);

            }
        };
        // 绑定队列
        channel.basicConsume(queueName, true, consumer);

        Thread.sleep(1000);
    }

    @Test
    public void dead_letter_queue_producer() throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();

        Channel channel = connection.createChannel();

        String bizExchangeName = "biz_exchange_2";
        String bizQueueName = "biz_queue_2";

        String deadExchangeName = "dead_exchange_2";
        String deadQueueName = "dead_queue_2";

        Map<String, Object> arguments = new HashMap<String, Object>(16);

        // 为队列设置队列交换机
        arguments.put("x-dead-letter-exchange", deadExchangeName);

        // 设置队列中的消息过期时间 ms
        arguments.put("x-message-ttl", 0);

        // 业务交换机声明
        channel.exchangeDeclare(bizExchangeName, "topic", false, false, null);
        // 业务队列声明
        channel.queueDeclare(bizQueueName, false, false, false, arguments);
        // 业务队列与交换机绑定
        channel.queueBind(bizQueueName, bizExchangeName, "biz.#");

        // 声明死信队列、交换机
        channel.exchangeDeclare(deadExchangeName, "topic", false, false, null);
        channel.queueDeclare(deadQueueName, false, false, false, null);
        channel.queueBind(deadQueueName, deadExchangeName, "biz.#");

        String message = "死信队列消息";
        channel.basicPublish(bizExchangeName, "biz.add", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

        System.out.println("success!!");
    }

    @Test
    public void dead_letter_queue_producer_2() throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();

        Channel channel = connection.createChannel();

        String bizExchangeName = "biz_exchange";
        String bizQueueName = "biz_queue";

        // 业务交换机声明
        channel.exchangeDeclare(bizExchangeName, "topic", false, false, null);
        /*// 业务队列声明
        channel.queu eDeclare
        (bizQueueName, false, false, false, null);
        // 业务队列与交换机绑定
        channel.queueBind(bizQueueName, bizExchangeName, "biz.#");*/

        String message = "死信队列消息";
        channel.basicPublish(bizExchangeName, "biz.add", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

        System.out.println("success!!");
    }

    @Test
    public void dead_letter_queue_consumer() throws IOException, TimeoutException, InterruptedException {

        Connection connection = ConnectionUtils.getConnection();

        Channel channel = connection.createChannel();

        String deadQueueName = "dead_queue";

        DefaultConsumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                System.out.println("死信队列收到消息。。。" + new String(body, "utf-8"));
                System.out.println("deliveryTag:" + envelope.getDeliveryTag());
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };

        // 监听死信队列，
        channel.basicConsume(deadQueueName, consumer);

        Thread.sleep(10000);
    }
}

/*
<configuration>
<group name="g1" default="true" thread="10">
<nodeRef name="52esseduxyy-test" />
</group>
<node name="52esseduxyy-test" uri="amqp://jzttxyyedu:jzOneEight0418edu@172.18.108.68:5672/testjz" address="172.18.108.68:5672" />
</configuration>*/
